package dht.simulation;

import java.util.HashMap;
import java.util.Vector;

import dht.data.Access;
import dht.data.AccessType;
import dht.data.Message;
import dht.data.MessageType;
import dht.statistics.DhtStatistics;

public class DataNode {

	public RouteTableV2 routeTable;
	public RouteTableV2 clientRouteTable;
	public int id;
	public int masterId;
	public double joinTime;
	public double idleTime;
	public double tableUpdateTime;
	public int tableVersion;
	// public boolean tableLock;
	public static final double diskTime = GlobalVariables.diskTime;

	public DataNode(int id, int masterId, double joinTime,
			double tableUpdateTime) {
		this.id = id;
		this.masterId = masterId;
		this.joinTime = joinTime;
		this.idleTime = this.joinTime;
		this.tableUpdateTime = tableUpdateTime;
		this.tableVersion = 0;
		// this.tableLock = false;
	}

	public void setRouteTable(RouteTableV2 routeTable) {
		if (routeTable.serverId2Version.get(this.id) >= this.tableVersion) {
			this.routeTable = routeTable;
			DhtStatistics.serverReport(id, 1, 0, 0, 0);
		} else {
			DhtStatistics.serverReport(id, 0, 1, 0, 0);
			// System.err.println("routeTable update rejected, serverId: " +
			// id);
		}
	}

	public void setClientRouteTable(RouteTableV2 routeTable) {
		this.clientRouteTable = routeTable;
		DhtStatistics.clientReport(this.id, 1);
	}

	private void cloneRouteTable() {
		this.routeTable = (RouteTableV2) this.routeTable.clone();
	}

	public Message serverStartAccess(Access access) throws DhtException {
		DhtStatistics.serverCpuReport(id, Math.max(idleTime, access.startTime)
				- idleTime, diskTime, Math.max(idleTime, access.startTime)
				+ diskTime);

		// if (tableLock) {
		// return null;
		// }
		// tableLock = true;

		Message msg = new Message();
		msg.accessId = access.accessId;
		msg.fromId = access.requesterId;
		msg.toId = masterId;
		msg.generateTime = access.startTime;
		if (access.accessType == AccessType.TABLE_UPDATE) {
			msg.msgType = MessageType.TABLE_REQ;
		}

		idleTime = Math.max(idleTime, msg.generateTime);
		msg.waitingCost = idleTime - msg.generateTime;
		msg.processCost = diskTime;
		msg.sendTime = msg.generateTime + msg.waitingCost + msg.processCost;
		idleTime = msg.sendTime;

		switch (msg.msgType) {
		case TABLE_REQ:
			msg.msgSize = GlobalVariables.tableReqSize;
			msg.table = routeTable;
			break;
		default:
			throw new DhtException("serverStartAccess msgType unknown");
		}
		return msg;
	}

	public Message clientStartAccess(Access access) throws DhtException {
		Message msg = new Message();
		msg.accessId = access.accessId;
		msg.fromId = access.requesterId;
		msg.toId = getEntry(access);
		msg.generateTime = access.startTime;
		if (access.accessType == AccessType.READ) {
			msg.msgType = MessageType.READ_REQ;
		} else if (access.accessType == AccessType.WRITE) {
			msg.msgType = MessageType.WRITE_REQ;
		}

		msg.waitingCost = 0;
		msg.processCost = diskTime;
		msg.sendTime = msg.generateTime + msg.waitingCost + msg.processCost;

		switch (msg.msgType) {
		case READ_REQ:
			msg.msgSize = GlobalVariables.readReqSize;
			break;
		case WRITE_REQ:
			msg.msgSize = GlobalVariables.writeReqSize + access.fileSize;
			break;
		default:
			throw new DhtException("clientStartAccess msgType unknown");
		}
		return msg;
	}

	public Vector<Message> serverProcessMsg(Message msg, Access access)
			throws DhtException {
		DhtStatistics.serverCpuReport(id,
				Math.max(idleTime, msg.getCurrentTime()) - idleTime, diskTime,
				Math.max(idleTime, msg.getCurrentTime()) + diskTime);
		if (msg.msgType == MessageType.READ_ACK
				|| msg.msgType == MessageType.WRITE_ACK
				|| msg.msgType == MessageType.READ_FWD_ACK
				|| msg.msgType == MessageType.WRITE_FWD_ACK
				|| msg.msgType == MessageType.SPLIT_ACK) {
			idleTime = Math.max(idleTime, msg.getCurrentTime()) + diskTime;
			return null;
		}
		Vector<Message> msgList = new Vector<Message>();
		Message newMsg = new Message();

		newMsg.accessId = msg.accessId;
		newMsg.fromId = msg.toId;
		newMsg.toId = msg.fromId;
		newMsg.generateTime = msg.getCurrentTime();

		idleTime = Math.max(idleTime, newMsg.generateTime);
		newMsg.waitingCost = idleTime - newMsg.generateTime;
		newMsg.processCost = diskTime;
		newMsg.sendTime = newMsg.generateTime + newMsg.waitingCost
				+ newMsg.processCost;
		idleTime = newMsg.sendTime;

		switch (msg.msgType) {
		case READ_REQ:
			if (routeTable.lookup(access.fileName) == id) {
				newMsg.msgSize = GlobalVariables.readRespSize + access.fileSize;
				newMsg.msgType = MessageType.READ_RESP;
				routeTable.increaseReadCounter(access.fileName, 1);
				DhtStatistics.serverReport(id, 0, 0, 1, 0);
			} else {
				newMsg.table = routeTable;
				newMsg.msgSize = GlobalVariables.readFwdSize;
				newMsg.msgType = MessageType.READ_FWD;
				newMsg.fwdToId = routeTable.lookup(access.fileName);
			}
			break;
		case WRITE_REQ:
			if (routeTable.lookup(access.fileName) == id) {
				newMsg.msgSize = GlobalVariables.writeRespSize;
				newMsg.msgType = MessageType.WRITE_RESP;
				routeTable.increaseWriteCounter(access.fileName, 1);
				DhtStatistics.serverReport(id, 0, 0, 0, 1);
			} else {
				newMsg.table = routeTable;
				newMsg.msgSize = GlobalVariables.writeFwdSize;
				newMsg.msgType = MessageType.WRITE_FWD;
				newMsg.fwdToId = routeTable.lookup(access.fileName);
			}
			break;
		case TABLE_RESP:
			setRouteTable(msg.table);
			newMsg.msgSize = GlobalVariables.tableAckSize;
			newMsg.msgType = MessageType.TABLE_ACK;
			// tableLock = false;
			break;
		case SPLIT_REQ:
			setRouteTable(msg.table);
			Message splitMoveMsg = (Message) newMsg.clone();
			splitMoveMsg.msgSize = GlobalVariables.splitMoveReqSize;
			splitMoveMsg.msgType = MessageType.SPLIT_MOVE_REQ;
			splitMoveMsg.waitingCost = newMsg.processCost;
			splitMoveMsg.processCost = diskTime;
			splitMoveMsg.sendTime = splitMoveMsg.generateTime
					+ splitMoveMsg.waitingCost + splitMoveMsg.processCost;
			splitMoveMsg.toId = msg.moveToId;
			splitMoveMsg.table = msg.table;
			splitMoveMsg.vecVnodes = getVnodes2Move(msg.moveToId,
					newMsg.sendTime);
			// TODO with idleTime
			msgList.add(splitMoveMsg);
			newMsg.msgSize = GlobalVariables.splitRespSize;
			newMsg.msgType = MessageType.SPLIT_RESP;
			break;
		case SPLIT_MOVE_REQ:
			setRouteTable(msg.table);
			cloneRouteTable();
			for (VNodeV2 vnode : msg.vecVnodes) {
				vnode.startTime = msg.getCurrentTime();
				routeTable.vnodeId2Vnode.put(vnode.id, (VNodeV2) vnode.clone());
			}
			++tableVersion;
			newMsg.vecVnodes = msg.vecVnodes;
			newMsg.msgSize = GlobalVariables.splitMoveRespSize;
			newMsg.msgType = MessageType.SPLIT_MOVE_RESP;
			break;
		case SPLIT_MOVE_RESP:
			cloneRouteTable();
			for (VNodeV2 vnode : msg.vecVnodes) {
				routeTable.vnodeId2Vnode.put(vnode.id, (VNodeV2) vnode.clone());
			}
			++tableVersion;
			newMsg.vecVnodes = msg.vecVnodes;
			newMsg.msgSize = GlobalVariables.splitMoveAckSize;
			newMsg.msgType = MessageType.SPLIT_MOVE_ACK;
			break;
		case SPLIT_MOVE_ACK:
			newMsg.vecVnodes = msg.vecVnodes;
			newMsg.toId = masterId;
			newMsg.moveSourceId = msg.fromId;
			newMsg.msgSize = GlobalVariables.splitFlushReqSize;
			newMsg.msgType = MessageType.SPLIT_FLUSH_REQ;
			break;
		case SPLIT_FLUSH_RESP:
			newMsg.msgSize = GlobalVariables.splitFlushAckSize;
			newMsg.msgType = MessageType.SPLIT_FLUSH_ACK;
			break;
		case READ_ACK:
			return null;
		case WRITE_ACK:
			return null;
		case READ_FWD_ACK:
			return null;
		case WRITE_FWD_ACK:
			return null;
		case SPLIT_ACK:
			return null;
		default:
			throw new DhtException("serverProcessMsg msgType unknown");
		}

		msgList.add(newMsg);
		return msgList;
	}

	public Vector<Message> clientProcessMsg(Message msg, Access access)
			throws DhtException {
		Vector<Message> msgList = new Vector<Message>();
		Message newMsg = new Message();

		newMsg.accessId = msg.accessId;
		newMsg.fromId = msg.toId;
		newMsg.toId = msg.fromId;
		newMsg.generateTime = msg.getCurrentTime();

		newMsg.waitingCost = 0;
		newMsg.processCost = diskTime;
		newMsg.sendTime = newMsg.generateTime + newMsg.waitingCost
				+ newMsg.processCost;
		Message fwdAckMsg;
		switch (msg.msgType) {
		case READ_FWD:
			// System.err.println("data read fwd from " + msg.fromId + " to "
			// + msg.fwdToId);
			fwdAckMsg = (Message) newMsg.clone();
			fwdAckMsg.msgSize = GlobalVariables.readFwdAckSize;
			fwdAckMsg.msgType = MessageType.READ_FWD_ACK;
			fwdAckMsg.waitingCost = newMsg.processCost;
			fwdAckMsg.processCost = diskTime;
			fwdAckMsg.sendTime = fwdAckMsg.generateTime + fwdAckMsg.waitingCost
					+ fwdAckMsg.processCost;
			msgList.add(fwdAckMsg);

			// newMsg.msgSize = GlobalVariables.metaReadReqSize;
			// newMsg.msgType = MessageType.META_READ_REQ;
			// newMsg.toId = GlobalVariables.masterId;

			setClientRouteTable(msg.table);
			newMsg.msgSize = GlobalVariables.readReqSize;
			newMsg.msgType = MessageType.READ_REQ;
			newMsg.toId = msg.fwdToId;
			break;
		case READ_RESP:
			newMsg.msgSize = GlobalVariables.readAckSize;
			newMsg.msgType = MessageType.READ_ACK;
			break;
		case META_READ_RESP:
			setClientRouteTable(msg.table);
			newMsg.msgSize = GlobalVariables.metaReadAckSize;
			newMsg.msgType = MessageType.META_READ_ACK;
			break;
		case WRITE_FWD:
			// System.err.println("data write fwd from " + msg.fromId + " to "
			// + msg.fwdToId);
			fwdAckMsg = (Message) newMsg.clone();
			fwdAckMsg.msgSize = GlobalVariables.writeFwdAckSize;
			fwdAckMsg.msgType = MessageType.WRITE_FWD_ACK;
			fwdAckMsg.waitingCost = newMsg.processCost;
			fwdAckMsg.processCost = diskTime;
			fwdAckMsg.sendTime = fwdAckMsg.generateTime + fwdAckMsg.waitingCost
					+ fwdAckMsg.processCost;
			msgList.add(fwdAckMsg);

			setClientRouteTable(msg.table);
			newMsg.msgSize = GlobalVariables.writeReqSize + access.fileSize;
			newMsg.msgType = MessageType.WRITE_REQ;
			newMsg.toId = msg.fwdToId;
			break;
		case WRITE_RESP:
			newMsg.msgSize = GlobalVariables.writeAckSize;
			newMsg.msgType = MessageType.WRITE_ACK;
			break;
		default:
			throw new DhtException("clientProcessMsg msgType unknown");
		}
		msgList.add(newMsg);
		return msgList;
	}

	private Vector<VNodeV2> getVnodes2Move(int moveToId, double currentTime) {
		Vector<VNodeV2> vecVnodes = new Vector<VNodeV2>();
		HashMap<Integer, Double> serverId2Load = routeTable.getServerLoads();
		double moveAmount = (serverId2Load.get(id) - serverId2Load
				.get(moveToId)) / 2;
		double addedAmount = 0.0;
		for (VNodeV2 vnode : routeTable.vnodeId2Vnode.values()) {
			if (vnode.serverId == id
					&& addedAmount + vnode.getLoad() < moveAmount) {
				addedAmount += vnode.getLoad();
				VNodeV2 newVnode = new VNodeV2(vnode.id, moveToId, currentTime);
				vecVnodes.add(newVnode);
			}
		}
		return vecVnodes;
	}

	private int getEntry(Access access) {
		int id;
		// id = NetworkUtil.getEntry(access.requesterId);
		id = clientRouteTable.lookup(access.fileName);
		return id;
	}
}
